package com.o2o.cleaning.month.platform.ebusiness_plat.jumei_2019_7.save_to_es

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark._
/**
  * @Description TODO 
  * @Author liutaowei
  * @Date 2018/12/3 14:56
  */
object jumei_to_247 {

  def main(args: Array[String]): Unit = {
    /***=============重要================================*/
    val platform_Name = "jumei"
    var month="11"
    var year = "2018"
    var nodes ="192.168.2.247"
    val obsFs = s"s3a://o2o-dataproces-group/"
    /***=============重要================================*/
    val spark = SparkSession.builder()
      .appName("SparkTest")
      .master("local[*]")
      .config("es.nodes", nodes)
      .config("es.port", "9200")
      .config("cluster.name","O2OElastic")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
      .getOrCreate()
//    val platform_shopUrl=obsFs + s"platdata/${year}/${month}/${platform_Name}/${platform_Name}_shop"

    val platform_goodUrl=obsFs + s"liu_taowei/month_data/platdata/2018/11/jumei/jumei_good/"

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "I6X0QOBRFQNDXGH6AMEU")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "uRrV8loOf6OszhzXz5GkkzBEQe5BY03vq5NEEWUw")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("WARN")



    val df =sc.textFile(platform_goodUrl)

    val value =df.map(line => {
      val lines=JSON.parseObject(line)
      val function = lines.getOrDefault("function","").toString
      //      val poor1 = lines.getOrDefault("poorCount","0").toString.replace("\"","")
      //      val generalCount = lines.getOrDefault("generalCount","0").toString.replace("\"","")
      //
      //      lines.put("poorCount",poor1)
      //      lines.put("generalCount",generalCount)
      lines.put("function",function)

      val nb = new JSONObject()

      nb.put("Base_Info",lines.getOrDefault("Base_Info","").toString)
      nb.put("address",lines.getOrDefault("address","").toString)
      nb.put("administrative_region",lines.getOrDefault("administrative_region","").toString)
      nb.put("allItemCount",lines.getOrDefault("allItemCount","").toString)
      nb.put("brandName",lines.getOrDefault("brandName","").toString)
      nb.put("brandName_cn",lines.getOrDefault("brandName_cn","").toString)
      nb.put("brandName_en",lines.getOrDefault("brandName_en","").toString)
      nb.put("brandValueId",lines.getOrDefault("brandValueId","").toString)
      nb.put("brand_isLaoZiHao",lines.getOrDefault("brand_isLaoZiHao","").toString)
      nb.put("brand_state",lines.getOrDefault("brand_state","").toString)
      nb.put("brand_type",lines.getOrDefault("brand_type","").toString)
      nb.put("categoryId",lines.getOrDefault("categoryId","").toString)
      nb.put("categoryName",lines.getOrDefault("categoryName","").toString)
      nb.put("city",lines.getOrDefault("city","").toString)
      nb.put("city_grade",lines.getOrDefault("city_grade","").toString)
      nb.put("city_origin",lines.getOrDefault("city_origin","").toString)
      nb.put("color",lines.getOrDefault("color","").toString)
      nb.put("commentCount",lines.getOrDefault("commentCount","").toString)
      nb.put("crawl_date",lines.getOrDefault("crawl_date","").toString)
      nb.put("district",lines.getOrDefault("district","").toString)
      nb.put("district_origin",lines.getOrDefault("district_origin","").toString)
      nb.put("economic_division",lines.getOrDefault("economic_division","").toString)
      nb.put("fans",lines.getOrDefault("fans","").toString)
      nb.put("favcount",lines.getOrDefault("favcount","").toString)
      nb.put("firstCategoryId",lines.getOrDefault("firstCategoryId","").toString)
      nb.put("fourthCategoryId",lines.getOrDefault("fourthCategoryId","").toString)
      nb.put("function",lines.getOrDefault("function","").toString)
      nb.put("generalCount",lines.getOrDefault("generalCount","").toString)
      nb.put("goodCount",lines.getOrDefault("goodCount","").toString)
      nb.put("goodRatePercentage",lines.getOrDefault("goodRatePercentage","").toString)
      nb.put("goodUrl",lines.getOrDefault("goodUrl","").toString)
      nb.put("good_id",lines.getOrDefault("good_id","").toString)
      nb.put("good_type",lines.getOrDefault("good_type","").toString)
      nb.put("if_city",lines.getOrDefault("if_city","").toString)
      nb.put("if_district",lines.getOrDefault("if_district","").toString)
      nb.put("if_state_level_new_areas",lines.getOrDefault("if_state_level_new_areas","").toString)
      nb.put("images",lines.getOrDefault("images","").toString)
      nb.put("is_onsell",lines.getOrDefault("is_onsell","").toString)
      nb.put("is_oversea",lines.getOrDefault("is_oversea","").toString)
      nb.put("is_selfSupport",lines.getOrDefault("is_selfSupport","").toString)
      nb.put("kaifaqu",lines.getOrDefault("kaifaqu","").toString)
      nb.put("model",lines.getOrDefault("model","").toString)
      nb.put("name",lines.getOrDefault("name","").toString)
      nb.put("original_cost",lines.getOrDefault("original_cost","").toString)
      nb.put("platformId",lines.getOrDefault("platformId","").toString)
      nb.put("platformName",lines.getOrDefault("platformName","").toString)
      nb.put("poorCount",lines.getOrDefault("poorCount","").toString)
      nb.put("poor_counties",lines.getOrDefault("poor_counties","").toString)
      nb.put("priceText",lines.getOrDefault("priceText","").toString)
      nb.put("province",lines.getOrDefault("province","").toString)
      nb.put("ram",lines.getOrDefault("ram","").toString)
      nb.put("regional_ID",lines.getOrDefault("regional_ID","").toString)
      nb.put("rom",lines.getOrDefault("rom","").toString)
      nb.put("rootCategoryId",lines.getOrDefault("rootCategoryId","").toString)
      nb.put("rootCategoryName",lines.getOrDefault("rootCategoryName","").toString)
      nb.put("rural_demonstration_counties",lines.getOrDefault("rural_demonstration_counties","").toString)
      nb.put("rural_ecommerce",lines.getOrDefault("rural_ecommerce","").toString)
      nb.put("salesAmount",lines.getOrDefault("salesAmount","").toString)
      nb.put("secondCategoryId",lines.getOrDefault("secondCategoryId","").toString)
      nb.put("sellCount",lines.getOrDefault("sellCount","").toString)
      nb.put("series",lines.getOrDefault("series","").toString)
      nb.put("shopId",lines.getOrDefault("shopId","").toString)
      nb.put("shopName",lines.getOrDefault("shopName","").toString)
      nb.put("shopType",lines.getOrDefault("shopType","").toString)
      nb.put("shopUrl",lines.getOrDefault("shopUrl","").toString)
      nb.put("subCategoryId",lines.getOrDefault("subCategoryId","").toString)
      nb.put("subCategoryName",lines.getOrDefault("subCategoryName","").toString)
      nb.put("the_belt_and_road_city",lines.getOrDefault("the_belt_and_road_city","").toString)
      nb.put("the_belt_and_road_province",lines.getOrDefault("the_belt_and_road_province","").toString)
      nb.put("the_yangtze_river_economic_zone_city",lines.getOrDefault("the_yangtze_river_economic_zone_city","").toString)
      nb.put("the_yangtze_river_economic_zone_province",lines.getOrDefault("the_yangtze_river_economic_zone_province","").toString)
      nb.put("thirdCategoryId",lines.getOrDefault("thirdCategoryId","").toString)
      nb.put("timeStamp",lines.getOrDefault("timeStamp","").toString)
      nb.put("title",lines.getOrDefault("title","").toString)
      nb.put("urban_agglomerations",lines.getOrDefault("urban_agglomerations","").toString)

      nb
    })

    value.saveToEs(s"2018_${platform_Name}_${month}/type_1"
      ,Map("es.mapping.id"->"good_id"))

    //    val df_shop =sc.textFile(platform_shopUrl)
    //    val value_shop =df_shop.map(line => {
    //      val lines=JSON.parseObject(line)
    //      lines
    //    })
    //    value_shop
    //      .saveToEs(s"2018_${platform_Name}_shop/${platform_Name}_shop_2018_${month}"
    //      ,Map("es.mapping.id"->"shopId"))
    //    val time = System.currentTimeMillis()
    //    val endtime = (System.currentTimeMillis()-time)/1000
  }
}
